package com.lovecaa.transaction;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 事务消息
 * 创建时间： 2023年10月26日 21:14:00
 *
 * @author Mr.Tao
 * @since 1.0.0
 */
public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionMQProducer producer = new TransactionMQProducer("TransactionProducer");
        producer.setNamesrvAddr("192.168.192.110:9876");
        // 设置本地事务监听器
        producer.setTransactionListener(new TransactionListenerImpl());
        // 核心线程数，最大线程数，存活时间，存活时间单位，堆积队列数
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5,
                100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000),
                r -> new Thread(r, "TransactionMessagePool"));
        // 异步提交事务提升性能
        producer.setExecutorService(threadPoolExecutor);
        producer.start();
        String[] tags = new String[]{"tagA", "tagB", "tagC", "tagD", "tagE"};
        for (int i = 0; i < 10; i++) {
            String topic = "TransactionTest";
            String tag = tags[i % tags.length];
            byte[] msg = (tag + "_TransactionProducer_" + i).getBytes(StandardCharsets.UTF_8);
            Message message = new Message(topic, tag, msg);
            TransactionSendResult result = producer.sendMessageInTransaction(message, "传递进去的参数");
            System.out.println("消息发送成功:" + result);
            Thread.sleep(10);
        }
    }
}
